/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */
package org.just.dusin.ooii.component.kj;

import gnu.io.*;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.util.*;
import java.util.logging.Level;
import javax.jms.*;
import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.RedeliveryPolicy;
import org.apache.log4j.Logger;
import org.dusin.webcoal.exception.OOIIException;
import org.just.dusin.ooii.component.Sink;
import org.just.dusin.ooii.component.Source;
import org.just.dusin.ooii.component.standard.AMQSource;
import org.just.dusin.ooii.database.DatabaseConnectionSQL;
import org.just.dusin.ooii.database.RealDeviceDao;

/**
 *
 * @author Administrator
 */
public class DayuanSpecialSource extends Source {

    private String brokerName = "broke";
    private static final Logger logger = Logger.getLogger(DayuanSpecialSource.class.getName());
    private int maxNoMessages = 5;
    private CollectorPollSource kjSource;
    private AMQSource amqSource;
    // private Source amqUploadSource; 
    private boolean isStarted = false;  //表示备机是否启动
    private boolean isBackup = false;  //当两边双闪时，是否主动停止。在网络通畅的 情况下，监测到另一台计算机发送来的消息时，主动停止本机
    private int maxReceivedFrom = 5;  //最多连续收到5条数据
    private int currentReceivedFrom=0;//当前在启动KJSource情况下,连续收到另外一台计算机数据的条数
    
    private Sink amqMainSink;
    private Thread kjThread = null;
    private Thread amqThreadForUpload=null;
    //private Thread uploadThread=null;
    private int waitForMainToLive = 10000; //等待主机启动时间
    private long MaxNoDataReceivedTime = 10000;//从采集器无数据的最大等待时间，超过此时间可认为主机停止工作    

    private RealDeviceDao realDevice=null;
    private DatabaseConnectionSQL dc = null;

    public DatabaseConnectionSQL getDc() {
        return dc;
    }

    public void setDc(DatabaseConnectionSQL dc) {
        this.dc = dc;
    }

    
    
    public RealDeviceDao getRealDevice() {
        return realDevice;
    }
    public void setRealDevice(RealDeviceDao realDevice) {
        this.realDevice = realDevice;
    }
    
    public int getMaxReceivedFrom() {
        return maxReceivedFrom;
    }

    public void setMaxReceivedFrom(int maxReceivedFrom) {
        this.maxReceivedFrom = maxReceivedFrom;
    }

    
    public boolean isIsBackup() {
        return isBackup;
    }

    public void setIsBackup(boolean isBackup) {
        this.isBackup = isBackup;
    }
    
    

    public long getMaxNoDataReceivedTime() {
        return MaxNoDataReceivedTime;
    }

    public void setMaxNoDataReceivedTime(long MaxNoDataReceivedTime) {
        this.MaxNoDataReceivedTime = MaxNoDataReceivedTime;
    }
    private long currentNoDataReceivedTime = 0;
    private long previousTimeofNoDataReceived = -1; //上次无数据的系统时间
    SimpleRead reader=null;
    int batchSize=10;

    public int getBatchSize() {
        return batchSize;
    }

    public void setBatchSize(int batchSize) {
        this.batchSize = batchSize;
    }

    
    public AMQSource getAmqSource() {
        return amqSource;
    }

    public void setAmqSource(AMQSource amqSource) {
        this.amqSource = amqSource;
    }
    
    
    public int getWaitForMainToLive() {
        return waitForMainToLive;
    }

    public void setWaitForMainToLive(int waitForMainToLive) {
        this.waitForMainToLive = waitForMainToLive;
    }

    public Sink getAmqMainSink() {
        return amqMainSink;
    }

    public void setAmqMainSink(Sink amqMainSink) {
        this.amqMainSink = amqMainSink;
    }

    /*
     * public Source getAmqUploadSource() { return amqUploadSource; }
     *
     * public void setAmqUploadSource(Source amqUploadSource) {
     * this.amqUploadSource = amqUploadSource;
    }
     */
    public CollectorPollSource getKjSource() {
        return kjSource;
    }

    public void setKjSource(CollectorPollSource kjSource) {
        this.kjSource = kjSource;
    }

    public int getMaxNoMessages() {
        return maxNoMessages;
    }

    public void setMaxNoMessages(int maxNoMessages) {
        this.maxNoMessages = maxNoMessages;
    }

    public String getBrokerName() {
        return brokerName;
    }

    public void setBrokerName(String brokerName) {
        this.brokerName = brokerName;
    }

    public String getBrokerURL() {
        return brokerURL;
    }

    public void setBrokerURL(String brokerURL) {
        this.brokerURL = brokerURL;
    }

    public String getPassword() {
        return password;
    }

    public void setPassword(String password) {
        this.password = password;
    }

    public String getQueueName() {
        return queueName;
    }

    public void setQueueName(String queueName) {
        this.queueName = queueName;
    }

    public String getUser() {
        return user;
    }

    public void setUser(String user) {
        this.user = user;
    }
    private String brokerURL = "tcp://localhost:61616";
    private String queueName = "coal";
    private int inteval = 300;

    public void setInteval(int inteval) {
        this.inteval = inteval;
    }
    private String user;
    private String password;

    @Override
    public void initComponent(Properties props) {
    }

    @Override
    public void destroyComponent() throws Exception {
    }

    private void setUplaodStatus(int s){
        String orgCode;
        orgCode="01010101";        
        try {
            realDevice = new RealDeviceDao(dc.getConnection());
            realDevice.updateSCStatus(s, orgCode);
        } catch (Exception ex) {
           logger.error("上传状态更新失败");
        }finally{
            dc.close();
        }
    }
    
    public void stopKJSource(){
        
         kjSource.stop();
        //amqUploadSource.stop();
        while ((kjThread != null && kjThread.isAlive())) {
            try {
                Thread.sleep(100);
            } catch (InterruptedException ex) {
                logger.warn("睡眠中断");
            }
        }
        setUplaodStatus(1);
        logger.info("备机采集线程停止");        
        isStarted = false;      
        reader =  this.new SimpleRead(kjSource.getPortId()); //启动采集器读数线程
    }
    
    public void startKJSource(){
        if(reader!=null){
            reader.setStop();
        }
        try {
            Thread.sleep(2000);
        } catch (InterruptedException ex) {
            logger.warn("睡眠中断");
        }
        if (kjSource.isStop()) {
            kjSource.setStop(false);
            kjThread = new Thread(kjSource);
            kjThread.start();
        } 
        if(amqSource.isStop()){
            amqSource.setStop(false);
            amqThreadForUpload = new Thread(amqSource);
            amqThreadForUpload.start();
        }
        isStarted = true;
    }
    
    @Override
    public void run() {
        ConnectionFactory factory = new ActiveMQConnectionFactory(brokerURL);
        int noMessageTimes = 0;
        reader =  this.new SimpleRead(kjSource.getPortId()); //启动采集器读数线程     
        setUplaodStatus(1);
        while (!stop) {
            //logger.info(kjSource.getSuccRate());
            Connection connection = null;
            Session session = null;
            MessageConsumer messageConsumer = null;
            try {
                if(previousTimeofNoDataReceived!=-1){
                    currentNoDataReceivedTime=System.currentTimeMillis()-previousTimeofNoDataReceived;
                }else{
                    previousTimeofNoDataReceived=System.currentTimeMillis();
                }
                connection = factory.createConnection(user, password);
                RedeliveryPolicy policy = ((ActiveMQConnection) connection).getRedeliveryPolicy();
                //policy.setRedeliveryDelay(100);
                policy.setMaximumRedeliveries(10);
                connection.start();
                //session.start();
                session = connection.createSession(true, Session.AUTO_ACKNOWLEDGE);
                Destination destination = session.createQueue(queueName);
                messageConsumer = session.createConsumer(destination);
                
                List<String> msgs = new ArrayList<String>();
                TextMessage txtMessage=null;
                int i=0;
                do{
                     txtMessage = (TextMessage) messageConsumer.receive(1000);
                     if(txtMessage!=null)
                         msgs.add(txtMessage.getText());
                     i++;
                }while(txtMessage!=null&&i<batchSize);
                
                if (msgs.size()>0) {
                    if(!isStarted){
                        noMessageTimes = 0;                               
                        logger.info("处理从主机来的消息:"+msgs.size()+"条");                   
                        sink.process(msgs);                    
                    }else{
                        logger.warn("舍弃主机来的消息:"+msgs.size()+"条");
                        if(msgs.size()==1){
                            currentReceivedFrom++;  //连续收到一条消息的数目
                            if(isBackup&&currentReceivedFrom>maxReceivedFrom){
                                logger.warn("监测到其他机器采集数据，停止本机采集");
                                stopKJSource();
                                currentReceivedFrom=0;
                            }
                        }else{
                            currentReceivedFrom=0;
                        }
                    }
                } else {
                    noMessageTimes++;
                    if (!isStarted && noMessageTimes > maxNoMessages&&currentNoDataReceivedTime>MaxNoDataReceivedTime) { //数次没有收到信息，则启动KJSource
                        //stop = true;
                        //启动另一个源组件;
                        logger.info("没有采集程序在运行，启动采集程序");
                        startKJSource();
                    }
                }
                session.commit();
                Thread.sleep(inteval);
            } catch (JMSException e) {
                if (session != null) {
                    try {
                        session.rollback();
                    } catch (JMSException ex) {
                        logger.error("事物回滚异常");
                    }
                }
                if (!isStarted) {
                    logger.warn("不能连接上主机，请检查主机备机网络通讯");
                    if(currentNoDataReceivedTime>MaxNoDataReceivedTime){
                        startKJSource();
                    }
                }

                logger.error("主机备机间无法通信，请检查网络连接");
            } catch (OOIIException e) {
                try {
                    session.rollback();
                } catch (JMSException ex) {
                    logger.error("事物回滚异常");
                }
                e.printStackTrace();
                logger.error(e.getMessage());
            } catch (InterruptedException e) {
                logger.error(e.getMessage());
            } catch(Exception e){
                e.printStackTrace();
                logger.error(e.getMessage());
            }finally {
                logger.info("关闭JMS连接");
                try {
                    try {
                        if (messageConsumer != null) {
                            messageConsumer.close();
                        }
                        if (session != null) {
                            session.close();
                        }
                        if (connection != null) {
                            connection.close();
                        }
                    } catch (JMSException e) {
                        logger.error("jms 关闭错误");
                    }

                    Thread.sleep(inteval);
                } catch (InterruptedException ex) {
                    logger.warn("睡眠中断");
                }
            }
        }
        setUplaodStatus(1);
        if(reader!=null){
            reader.setStop();
        }
    }
    
    
    
    public class SimpleRead implements Runnable, SerialPortEventListener {   //内部

        CommPortIdentifier portId;
        Enumeration portList;
        InputStream inputStream;
        OutputStream outputStream;
        SerialPort serialPort;
        Thread readThread;
        private boolean stop = false;

        public SimpleRead(String portName) {
            try {
                portId = CommPortIdentifier.getPortIdentifier(portName);
                serialPort = (SerialPort) portId.open("SimpleReadApp", 2000);
            } catch (PortInUseException e) {
                logger.error(e.getMessage());
            } catch (NoSuchPortException e) {
                logger.error(e.getMessage());
            }
            try {
                outputStream = serialPort.getOutputStream();
                inputStream = serialPort.getInputStream();
            } catch (IOException e) {
                logger.error(e.getMessage());
            }
            try {
                serialPort.addEventListener(this);
            } catch (TooManyListenersException e) {
                logger.error(e.getMessage());
            }
            serialPort.notifyOnDataAvailable(true);

            readThread = new Thread(this);
            logger.info("启动采集器数据接受线程");
            readThread.start();            
        }

        public void run() {
            while (!stop) {
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                }
            }
            try {
                inputStream.close();
                outputStream.close();
                serialPort.close();
            } catch (IOException e) {
                logger.error(e.getMessage());
            }
            logger.info("结束采集器数据接收线程");
        }

        public void serialEvent(SerialPortEvent event) {
            switch (event.getEventType()) {
                case SerialPortEvent.BI:
                case SerialPortEvent.OE:
                case SerialPortEvent.FE:
                case SerialPortEvent.PE:
                case SerialPortEvent.CD:
                case SerialPortEvent.CTS:
                case SerialPortEvent.DSR:
                case SerialPortEvent.RI:
                case SerialPortEvent.OUTPUT_BUFFER_EMPTY:
                    break;
                case SerialPortEvent.DATA_AVAILABLE:                                
                    currentNoDataReceivedTime = 0;
                    previousTimeofNoDataReceived = System.currentTimeMillis();
                    
                    //System.out.println(diffDataTime);
                    try {
                        Thread.sleep(500);
                    } catch (InterruptedException ex) {
                    }                    
                    
                    byte[] readBuffer = new byte[92];
                    try {
                        while (inputStream.available() > 0) {
                            int numBytes = inputStream.read(readBuffer);
                            logger.info(numBytes + " bytes received");
                        }                      
                    } catch (IOException e) {
                    }
                    break;
            }
        }
        
        public void setStop(){
            stop=true;
        }
        public void setStart(){
            stop=false;
        }
    }
}